package com.zzy.controller;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.common.network.Selector;
/**
 * 测试 发送消息
 */
public class ProducerSend {


    public static void main(String[] args) {
        try {
            String serverList = "192.168.21.131:9092";
            String topic = "topic_test";
            Properties prop = new Properties();
            //2.给Kafka配置对象添加配置信息
            // 配置bootstrap.servers
            prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverList);
            //配置key.serializer 和value.serializer
            prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

            prop.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
            prop.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024 );

            //创建Kafka生产者对象
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(prop);
            // 调用send 方法，发送消息
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println("1111111");
            for(int i=5;i>0;i--){
                System.out.println("send:-->" + i );
                String v = "demo_zzy_"+ i+ "_" + sdf.format(new Date());
                kafkaProducer.send
                        (new ProducerRecord<>(topic, v )).get();
            }


//            Future<RecordMetadata> future = kafkaProducer.send
//                    (new ProducerRecord<>(topic, "demo_zzy_" + sdf.format(new Date())));
//            System.out.println("22222");
//            RecordMetadata recordMetadata = future.get();
//            System.out.println(333333);
//            System.out.println("同步发送消息结果：" + "topic" + recordMetadata.topic() + "partition" + recordMetadata.partition()
//                    + "offset" + recordMetadata.offset());

        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
